/*
Copyright (C) 2010- Peter Bui and Aaron Dingler.
This software is distributed under the GNU General Public License.
See the file COPYING for details.
*/

#include "smq_common.h"
#include "smq_message.h"
#include "smq_message_queue.h"
#include "smq_util.h"

#include <create_dir.h>
#include <delete_dir.h>
#include <debug.h>
#include <hash_table.h>
#include <int_sizes.h>
#include <link.h>
#include <list.h>
#include <sort_dir.h>
#include <stringtools.h>
#include <xmalloc.h>

#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>

/* Globals ------------------------------------------------------------------*/

static struct hash_table *SMQ_Exchange_Log_Table = NULL;
static struct list* SMQ_Exchange_Sentinel_List = NULL;
static FILE *SMQ_Exchange_Log_Stream = NULL;

/* Internal functions -------------------------------------------------------*/

#define SMQ_EXCHANGE_INSERT_SENTINEL(list, key) \
    list = hash_table_remove(SMQ_Exchange_Log_Table, (key)); \
    if (list && list != SMQ_Exchange_Sentinel_List) { \
	list_free(list); \
	list_delete(list); \
    } \
    hash_table_insert(SMQ_Exchange_Log_Table, (key), SMQ_Exchange_Sentinel_List);

void
smq_exchange_setup_log(const char *root, const char *name)
{
    char log_path[SMQ_LINE_MAX];
    char meta_path[SMQ_LINE_MAX];
    char body_path[SMQ_LINE_MAX];
    char line[SMQ_LINE_MAX];
    char *message;
    struct list *bindings;
    int result;

    if (SMQ_Exchange_Log_Table) {
	hash_table_firstkey(SMQ_Exchange_Log_Table);
	while (hash_table_nextkey(SMQ_Exchange_Log_Table, &message, (void **)&bindings)) {
	    list_free(bindings);
	    list_delete(bindings);
	}

	hash_table_delete(SMQ_Exchange_Log_Table);
    }
    SMQ_Exchange_Log_Table = hash_table_create(0, 0);

    snprintf(log_path, SMQ_LINE_MAX, "%s/%s/.__exchange.log", root, name);
    if (SMQ_Exchange_Log_Stream) 
	fclose(SMQ_Exchange_Log_Stream);

    if (!(SMQ_Exchange_Log_Stream = fopen(log_path, "a+"))) 
	fatal("unable to open header log %s: %s", log_path, strerror(errno));

    while (fgets(line, SMQ_LINE_MAX, SMQ_Exchange_Log_Stream)) {
	char **argv;
	int argc;

	string_chomp(line);

	if (!string_split_quotes(line, &argc, &argv)) {
	    debug(D_NOTICE, "unable to split exchange log %s: %s", line, strerror(errno));
	    continue;
	}

	if (argc == 0) {
	    free(argv);
	    continue;
	}

	if (strcmp(argv[0], "bind") == 0 && argc == 3) {
	    snprintf(meta_path, SMQ_LINE_MAX, "%s/%s/%s.meta", root, name, argv[1]);
	    if (!smq_util_file_exists(meta_path)) continue;
	    
	    bindings = hash_table_lookup(SMQ_Exchange_Log_Table, argv[1]);
	    if (!bindings) {
		bindings = list_create();
		hash_table_insert(SMQ_Exchange_Log_Table, argv[1], bindings);
	    }

	    list_push_tail(bindings, (void *)(PTRINT_T)xstrdup(argv[2]));
	    debug(D_DEBUG, "mark bind %s %s", argv[1], argv[2]);
	} else if (strcmp(argv[0], "done") == 0 && argc == 2) {	// TODO: change to bound
	    SMQ_EXCHANGE_INSERT_SENTINEL(bindings, argv[1]);
	    debug(D_DEBUG, "mark %s done", argv[1]);
	} else if (strcmp(argv[0], "flush") == 0 && argc == 2) {
	    result = 0;

	    snprintf(meta_path, SMQ_LINE_MAX, "%s/%s/%s.meta", root, name, argv[1]);
	    if (smq_util_file_exists(meta_path)) 
		result += unlink(meta_path);

	    snprintf(body_path, SMQ_LINE_MAX, "%s/%s/%s.body", root, name, argv[1]);
	    if (smq_util_file_exists(body_path)) 
		result += unlink(body_path);

	    if (result == 0) {
		bindings = hash_table_remove(SMQ_Exchange_Log_Table, argv[1]);
		if (bindings && bindings != SMQ_Exchange_Sentinel_List) {
		    list_free(bindings);
		    list_delete(bindings);
		}
	    } else {
		SMQ_EXCHANGE_INSERT_SENTINEL(bindings, argv[1]);
	    }
	    debug(D_DEBUG, "mark %s flush", argv[1]);
	} else {
	    debug(D_NOTICE, "unknown exchange log entry: %s", line);
	}

	free(argv);
    }
}

static void
smq_exchange_run_bindings(const char *queue, const char *mid)
{
    int i;
    int status;
    struct list *bindings;
    char *binding;
    char **binding_scripts; 
    char bindings_path[SMQ_LINE_MAX];
    char sandbox_path[SMQ_LINE_MAX];
    char command[SMQ_LINE_MAX];
    int all_success = 1;

    bindings = hash_table_lookup(SMQ_Exchange_Log_Table, mid);
    if (bindings == SMQ_Exchange_Sentinel_List) return;
    if (bindings == NULL) {
	bindings = list_create();
	hash_table_insert(SMQ_Exchange_Log_Table, mid, bindings);
    }
	
    snprintf(bindings_path, SMQ_LINE_MAX, "%s/.bindings", queue);
    snprintf(sandbox_path, SMQ_LINE_MAX, "%s/.sandbox", queue);

    if (!sort_dir(bindings_path, &binding_scripts, strcmp)) {
	debug(D_NOTICE, "unable to get sorted list of binding scripts %s: %s", bindings_path, strerror(errno));
	return;
    }
	
    debug(D_DEBUG, "run bindings on %s/%s", queue, mid);
    for (i = 0; binding_scripts[i]; i++) {
	if (binding_scripts[i][0] == '.') continue;

	list_first_item(bindings);
	while ((binding = list_next_item(bindings)))
	    if (strcmp(binding_scripts[i], binding) == 0)
		continue;

	if (!create_dir(sandbox_path, 0700)) 
	    fatal("unable to create sandbox directory %s: %s", sandbox_path, strerror(errno));

	snprintf(command, SMQ_LINE_MAX, "../.bindings/%s %s/%s.meta %s/%s.body", binding_scripts[i], queue, mid, queue, mid);
	chdir(sandbox_path);
	status = system(command);
	if (WIFEXITED(status)) {
	    debug(D_DEBUG, "binding %s on %s/%s = %d", binding_scripts[i], queue, mid, WEXITSTATUS(status));
	    if (WEXITSTATUS(status) == 0) {
		fprintf(SMQ_Exchange_Log_Stream, "bind %s %s\n", mid, binding_scripts[i]);
		fflush(SMQ_Exchange_Log_Stream);
		list_push_tail(bindings, xstrdup(binding_scripts[i]));
	    }
	} else {
	    debug(D_NOTICE, "binding %s on %s/%s = %d", binding_scripts[i], queue, mid, status);
	    all_success = 0;
	}
	chdir(queue);
	delete_dir(sandbox_path);
    }
    
    sort_dir_free(binding_scripts);

    if (all_success && list_size(bindings) > 0) {
	fprintf(SMQ_Exchange_Log_Stream, "done %s\n", mid); // TODO: replace w/ bound
	fflush(SMQ_Exchange_Log_Stream);
	SMQ_EXCHANGE_INSERT_SENTINEL(bindings, mid);
    }
}

static void
smq_exchange_flush(const char *queue, const char *mid)
{
    struct list *bindings;
    char meta_path[SMQ_LINE_MAX];
    char body_path[SMQ_LINE_MAX];
    
    bindings = hash_table_lookup(SMQ_Exchange_Log_Table, mid);
    if (bindings != SMQ_Exchange_Sentinel_List) return;

    debug(D_DEBUG, "flushing %s", mid);
    snprintf(meta_path, SMQ_LINE_MAX, "%s/%s.meta", queue, mid);
    snprintf(body_path, SMQ_LINE_MAX, "%s/%s.body", queue, mid);
    
    fprintf(SMQ_Exchange_Log_Stream, "flush %s\n", mid);
    fflush(SMQ_Exchange_Log_Stream);
    
    unlink(meta_path);
    unlink(body_path);

    hash_table_remove(SMQ_Exchange_Log_Table, mid);
}

/* Public functions ---------------------------------------------------------*/

void
smq_exchange_run(const char *root, const char *name, int timeout)
{
    char queue[SMQ_LINE_MAX];
    char flush[SMQ_LINE_MAX];
    char mid[SMQ_LINE_MAX];
    char **message_list;
    int do_flush;

    SMQ_Exchange_Sentinel_List = list_create();

    smq_exchange_setup_log(root, name);

    snprintf(queue, SMQ_LINE_MAX, "%s/%s", root, name);
    snprintf(flush, SMQ_LINE_MAX, "%s/.__flush", queue);
    setenv("SMQ_EXCHANGE_QUEUE", queue, 1);

    while (1) {
	int i;

	/* TODO: compact log after a certain amount of time */
	/* TODO: garbage collect */
	
	sleep(timeout);

	if (!sort_dir(queue, &message_list, strcmp)) {
	    debug(D_NOTICE, "unable to get sorted list of queue %s: %s", queue, strerror(errno));
	    continue;
	}
		
	do_flush = smq_util_file_exists(flush);

	for (i = 0; message_list[i]; i++) {
	    if (strstr(message_list[i], ".meta")) {
		strncpy(mid, message_list[i], strlen(message_list[i]) - 5);
		mid[strlen(message_list[i]) - 5] = 0;

		if (do_flush)
		    smq_exchange_flush(queue, mid);
		else
		    smq_exchange_run_bindings(queue, mid);
	    }
	}

	sort_dir_free(message_list);

	if (do_flush && unlink(flush) < 0)
	    fatal("unable to remove flush file %s: %s", flush, strerror(errno));
    }
}

/* 
vim: sw=4 sts=4 ts=8 ft=cpp
*/
